package Demo1;

import java.io.IOException;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

public class DemoBlotPrint implements IRichBolt{
	static Socket _socket=null;
	PrintWriter _out=null;
	
	@Override
	public void cleanup() {
		try{
			_socket.close();
			_out.close();
		}catch(IOException e){}
	}

	@Override
	public void execute(Tuple arg0) {
		if(_socket==null){
			try {
				_socket=new Socket("192.168.181.124",8002);
				_out = new PrintWriter(_socket.getOutputStream());
			} catch (UnknownHostException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} 
		}
		_out.write(arg0.getString(0));
		_out.flush();
	}

	@Override
	public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
		if(_socket==null){
			try {
				_socket=new Socket("192.168.181.124",8002);
				_out = new PrintWriter(_socket.getOutputStream());
			} catch (UnknownHostException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} 
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}
	
}